package com.speedfall.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class Produce {
    private Producer<String,String> producer;
    private Properties props;

    public void init() {
        props = new Properties();
        /*props.put("bootstrap.servers", "192.168.216.138:9092,192.168.216.139:9092,192.168.216.140:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");*/
        InputStream in = null;
        try {
            ClassLoader classLoader = this.getClass().getClassLoader();
            in = classLoader.getResourceAsStream("parameter.properties");
            props.load(in);
            producer = new KafkaProducer<String,String>(props);
        } catch (IOException e) {
        }finally {
            try {
                in.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public void sendKafaMessage(String topic,String message) {
        init();
        producer.send(new ProducerRecord<String, String>(topic, message));
        producer.close();
        System.out.println("finish");
    }

    public static void main(String[] args) {
        new Produce().sendKafaMessage("speedfall","hello world");
    }
}
